package com.zeroz.zookeeper.curator;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.atomic.AtomicValue;
import org.apache.curator.framework.recipes.atomic.DistributedAtomicInteger;
import org.apache.curator.framework.recipes.barriers.DistributedBarrier;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * Curator高级功能示例
 * 展示Curator提供的分布式协调工具
 */
public class CuratorRecipesExample {
    private static final String CONNECT_STRING = "localhost:2181,localhost:2182,localhost:2183";
    private static final int SESSION_TIMEOUT = 5000;
    private static final int CONNECTION_TIMEOUT = 5000;
    private CuratorFramework client;
    
    /**
     * 连接到ZooKeeper
     */
    public void connect() {
        // 创建重试策略
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        
        // 创建Curator客户端
        client = CuratorFrameworkFactory.builder()
                .connectString(CONNECT_STRING)
                .sessionTimeoutMs(SESSION_TIMEOUT)
                .connectionTimeoutMs(CONNECTION_TIMEOUT)
                .retryPolicy(retryPolicy)
                .namespace("recipes") // 所有操作都在/recipes命名空间下
                .build();
        
        // 启动客户端
        client.start();
        System.out.println("Curator客户端已启动");
    }
    
    /**
     * 关闭连接
     */
    public void close() {
        if (client != null) {
            client.close();
            System.out.println("Curator客户端已关闭");
        }
    }
    
    /**
     * 分布式锁示例
     */
    public void distributedLockDemo() throws Exception {
        final String lockPath = "/locks/lock_demo";
        final InterProcessMutex lock = new InterProcessMutex(client, lockPath);
        
        System.out.println("尝试获取分布式锁...");
        if (lock.acquire(10, TimeUnit.SECONDS)) {
            try {
                System.out.println("成功获取锁！执行受保护的代码...");
                // 模拟执行一些需要加锁的操作
                Thread.sleep(2000);
            } finally {
                System.out.println("释放锁...");
                lock.release();
            }
        } else {
            System.out.println("无法获取锁!");
        }
    }
    
    /**
     * 分布式计数器示例
     */
    public void distributedCounterDemo() throws Exception {
        String counterPath = "/counters/demo_counter";
        
        // 创建分布式计数器
        DistributedAtomicInteger counter = new DistributedAtomicInteger(
                client, counterPath, new ExponentialBackoffRetry(1000, 3));
        
        System.out.println("初始化计数器...");
        AtomicValue<Integer> value = counter.get();
        System.out.println("当前计数: " + value.preValue());
        
        // 增加计数
        System.out.println("增加计数...");
        value = counter.increment();
        if (value.succeeded()) {
            System.out.println("增加成功! 旧值: " + value.preValue() + ", 新值: " + value.postValue());
        }
        
        // 再次增加计数
        value = counter.add(5);
        if (value.succeeded()) {
            System.out.println("增加5成功! 旧值: " + value.preValue() + ", 新值: " + value.postValue());
        }
    }
    
    /**
     * 领导选举示例
     */
    public void leaderElectionDemo() throws Exception {
        final String leaderPath = "/leader/election_demo";
        final CountDownLatch leaderLatch = new CountDownLatch(1);
        
        LeaderSelector leaderSelector = new LeaderSelector(client, leaderPath,
                new LeaderSelectorListenerAdapter() {
                    @Override
                    public void takeLeadership(CuratorFramework client) throws Exception {
                        // 当前客户端被选为leader
                        System.out.println("当前客户端被选为Leader");
                        
                        try {
                            // 模拟执行领导工作
                            System.out.println("开始执行Leader工作");
                            Thread.sleep(5000);
                            System.out.println("完成Leader工作");
                        } finally {
                            System.out.println("放弃领导权");
                            leaderLatch.countDown();
                        }
                    }
                });
        
        // 确保在故障后还能再次获得领导权
        leaderSelector.autoRequeue();
        
        // 启动leader选举
        leaderSelector.start();
        System.out.println("启动Leader选举");
        
        // 等待领导工作完成
        leaderLatch.await();
        
        // 关闭leader选举器
        leaderSelector.close();
        System.out.println("关闭Leader选举器");
    }
    
    /**
     * 子节点监听示例
     */
    public void childrenCacheDemo() throws Exception {
        final String watchPath = "/watch/children";
        
        // 确保路径存在
        if (client.checkExists().forPath(watchPath) == null) {
            client.create().creatingParentsIfNeeded().forPath(watchPath, "parent".getBytes());
        }
        
        // 创建PathChildrenCache来监听子节点变化
        PathChildrenCache childrenCache = new PathChildrenCache(client, watchPath, true);
        
        // 添加监听器
        childrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                switch (event.getType()) {
                    case CHILD_ADDED:
                        System.out.println("子节点添加: " + event.getData().getPath());
                        break;
                    case CHILD_UPDATED:
                        System.out.println("子节点更新: " + event.getData().getPath());
                        break;
                    case CHILD_REMOVED:
                        System.out.println("子节点删除: " + event.getData().getPath());
                        break;
                    default:
                        break;
                }
            }
        });
        
        // 启动缓存
        childrenCache.start();
        System.out.println("子节点监听已启动");
        
        // 创建子节点以触发事件
        System.out.println("创建子节点...");
        String child1Path = watchPath + "/child1";
        client.create().forPath(child1Path, "child1 data".getBytes());
        
        // 稍等片刻以便监听器处理
        Thread.sleep(1000);
        
        // 更新子节点
        System.out.println("更新子节点...");
        client.setData().forPath(child1Path, "updated child1 data".getBytes());
        
        // 稍等片刻以便监听器处理
        Thread.sleep(1000);
        
        // 删除子节点
        System.out.println("删除子节点...");
        client.delete().forPath(child1Path);
        
        // 稍等片刻以便监听器处理
        Thread.sleep(1000);
        
        // 关闭缓存
        childrenCache.close();
        System.out.println("子节点监听已关闭");
    }
    
    /**
     * 分布式屏障示例
     */
    public void distributedBarrierDemo() throws Exception {
        final String barrierPath = "/barrier/demo";
        
        // 创建分布式屏障
        DistributedBarrier barrier = new DistributedBarrier(client, barrierPath);
        
        // 设置屏障
        barrier.setBarrier();
        System.out.println("屏障已设置");
        
        // 模拟多个线程等待屏障
        Thread waiterThread = new Thread(() -> {
            try {
                System.out.println("线程正在等待屏障释放...");
                barrier.waitOnBarrier();
                System.out.println("屏障已释放，线程继续执行!");
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
        
        // 启动等待线程
        waiterThread.start();
        
        // 主线程等待一段时间后释放屏障
        Thread.sleep(5000);
        System.out.println("准备释放屏障...");
        barrier.removeBarrier();
        
        // 等待线程完成
        waiterThread.join();
    }
    
    public static void main(String[] args) {
        CuratorRecipesExample example = new CuratorRecipesExample();
        
        try {
            // 连接到ZooKeeper
            example.connect();
            
            // 演示分布式锁
            System.out.println("\n===== 分布式锁示例 =====");
            example.distributedLockDemo();
            
            // 演示分布式计数器
            System.out.println("\n===== 分布式计数器示例 =====");
            example.distributedCounterDemo();
            
            // 演示Leader选举
            System.out.println("\n===== Leader选举示例 =====");
            example.leaderElectionDemo();
            
            // 演示子节点监听
            System.out.println("\n===== 子节点监听示例 =====");
            example.childrenCacheDemo();
            
            // 演示分布式屏障
            System.out.println("\n===== 分布式屏障示例 =====");
            example.distributedBarrierDemo();
            
        } catch (Exception e) {
            System.err.println("发生错误: " + e.getMessage());
            e.printStackTrace();
        } finally {
            example.close();
        }
    }
} 